package com.demo.feifan.consumer;

import com.demo.feifan.entity.Test1;
import com.demo.feifan.entity.Test2;
import com.demo.feifan.mapperstruct.TestMapperConver;
import com.demo.feifan.service.TestService;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.JsonDeserializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;

@Component
@RequiredArgsConstructor
public class TestConsumer {
    private static final Logger logger = LoggerFactory.getLogger(TestConsumer.class);
    private static final int BATCH_SIZE = 2000;

    // 为业务数据创建独立的缓冲区和锁
    private final List<String> buffer = new ArrayList<>();
    private final ReentrantLock bufferLock = new ReentrantLock();


    private final TestService testService;

    private final TestMapperConver testMapperConver;

    private final ObjectMapper objectMapper = new ObjectMapper()
            .registerModule(new JavaTimeModule())
            .setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"))
            .registerModule(new SimpleModule()
                    .addDeserializer(LocalDateTime.class, new JsonDeserializer<LocalDateTime>() {
                        @Override
                        public LocalDateTime deserialize(JsonParser p, DeserializationContext ctxt)
                                throws IOException {
                            String dateStr = p.getText();
                            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                            return LocalDateTime.parse(dateStr, formatter);
                        }
                    }));



    @KafkaListener(topicPartitions = {
            @TopicPartition(topic = "T-Demo", partitionOffsets = {
                    @PartitionOffset(partition = "0", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "1", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "2", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "3", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "4", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "5", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "6", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "7", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "8", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "9", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "10", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "11", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "12", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "13", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "14", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "15", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "16", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "17", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "18", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "19", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "20", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "21", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "22", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "23", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "24", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "25", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "26", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "27", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "28", initialOffset = Long.MAX_VALUE + ""),
                    @PartitionOffset(partition = "29", initialOffset = Long.MAX_VALUE + "")
            })
    })
    public void consumeTaxiGps(String message) throws JsonProcessingException {
        try {
            bufferLock.lock();
            buffer.add(message);

            if (buffer.size() >= BATCH_SIZE) {
                List<String> batchToSave = new ArrayList<>(buffer);
                buffer.clear();
                bufferLock.unlock(); // 提前释放锁
                saveBatchToDB(batchToSave);
            }
        } catch (Exception e) {
            logger.error("处理消息时出错: {}", message, e);
        } finally {
            if (bufferLock.isHeldByCurrentThread()) {
                bufferLock.unlock();
            }
        }
    }


    private void saveBatchToDB(List<String> batch) {
        try {
            List<Test1> gpsList = new ArrayList<>();

            // 将缓冲区中的消息转换为 CommonTaxiGps 对象
            for (String message : batch) {
                try {
                    // 1. 将 String 转换为 test2 对象
                    Test2 test2 = objectMapper.readValue(message, Test2.class);

                    // 使用mapstruct映射器将   没什么用 告诉下你可以用mapstruct转化下  当初我需要转换使用做逻辑
                    Test1 test1 = testMapperConver.toTest1(test2);
                    gpsList.add(test1);
                } catch (Exception e) {
                    logger.error("解析消息时出错: {}", message, e);
                }
            }
            // 使用 批量插入数据
            if (!gpsList.isEmpty()) {

                testService.insertBatch(gpsList);
                logger.info("成功批量插入 {} 条记录到数据库", gpsList.size());
            }
        } catch (Exception e) {
            logger.error("批量保存到数据库时出错", e);
        }
    }

}
